#pragma once
#include <sys/epoll.h>
#include "Log.hpp"
#include <string.h>
#include <errno.h>
#include <unordered_map>
#include <memory>
#include <vector>
#include <functional>
#include <sys/eventfd.h>
#include "Socket2.hpp"
#include <thread>
#include <condition_variable>
#include "Buffer.hpp"
// #include "EventLoop.hpp"
using namespace LogMoudle;
#define MAX_EPOLL_SIZE 1024
class EventLoop;
class Poller;
using EventCallback = std::function<void(void)>;
using PollerPtr = std::shared_ptr<Poller>;
#define InBufferSize (1024 * 10 - 1)
#define DEAFAULT_THREAD_COUNT 16

class Channel
{
private:
    uint32_t _event;           // 当前监控室=事件
    uint32_t _triggered_event; // 被触发的事件
    int _fd;
    // Poller& _poller;
    EventLoop &_event_loop;
    EventCallback _read_callback;
    EventCallback _write_callback;
    EventCallback _close_callback;
    EventCallback _error_callback;
    EventCallback _event_callback;

public:
    Channel(int fd, EventLoop &event_loop) : _fd(fd),
                                             _triggered_event(0),
                                             _event(0),
                                             _event_loop(event_loop)
    //_poller(poller)
    {
        _event = _event | EPOLLHUP;
        _event = _event | EPOLLPRI;
        _event = _event | EPOLLERR;
        _event = _event | EPOLLHUP;
    }

    void Init(int fd)
    {
        _fd = fd;
        _triggered_event = 0;
    }
    void Update();
#pragma region 设置回调
    void SetReadCallback(const EventCallback &cb) { _read_callback = cb; }
    void SetWriteCallback(const EventCallback &cb) { _write_callback = cb; }
    void SetCloseCallback(const EventCallback &cb) { _close_callback = cb; }
    void SetErrorCallback(const EventCallback &cb) { _error_callback = cb; }
    void SetEventCallback(const EventCallback &cb) { _event_callback = cb; }
#pragma endregion
    bool isReadable() { return _event & EPOLLIN; }   // EPOLLIN可读
    bool isWriteable() { return _event & EPOLLOUT; } // EPOLLOUT可写 不可以在可写的基础上再次修改为可写epoll特性

    void EnableRead()
    {
        _event |= EPOLLIN;
        // Update(*this);
        Update();
    }
    void EnableWrite()
    {
        _event |= EPOLLOUT;
        // Update(*this);
        Update();
    }
    void DisableRead()
    {
        _event &= ~EPOLLIN;
        // Update();
        Update();
    }
    void DisableWrite()
    {
        _event &= ~EPOLLOUT;
        // Update();
        Update();
    }
    void DisableAll()
    {
        _event = 0;
        // Update();
        Update();
    }

    void Remove();
    void SetTriggeredEvent(uint32_t event) { _triggered_event = event; }
    void HandleEvent() // 处理事件
    {
        if ((_triggered_event & EPOLLIN) || (_triggered_event & EPOLLHUP) || (_triggered_event & EPOLLPRI)) // 读事件 对方关闭 优先带外数据
        {

            if (_read_callback)
                _read_callback();
        }
        // std::cout << __LINE__ << " : " << Fd() << _triggered_event<< std::endl;
        if (_triggered_event & EPOLLOUT)
        {
            if (_write_callback)
                _write_callback();
        }
        else if (_triggered_event & EPOLLHUP)
        {
            LOG(LogLevel::DEBUG) << "EPOLLHUP";
            if (_close_callback)
                _close_callback();
        }
        else if (_triggered_event & EPOLLERR)
        {
            LOG(LogLevel::DEBUG) << "EPOLLERR";
            if (_error_callback)
                _error_callback();
        }

        if (_event_callback) // 刷新活跃度使业务处理时间不算在活跃度内
            _event_callback();
    }

    int Fd() { return _fd; }
    uint32_t Event() { return _event; }
};
typedef std::shared_ptr<Channel> ChannelPtr;
class Poller
{

private:
    int _epollfd;
    std::unordered_map<int, Channel *> _channels; // 给每个进入我路线记起来
    /// @brief 调用epoll_ctl更改监控事件
    /// @param channel
    /// @param op
    void Update(Channel *channel, int op)
    {

        epoll_event event; // 创建epoll_ctl必须参数
        event.data.fd = channel->Fd();
        event.events = channel->Event();
        if (epoll_ctl(_epollfd, op, channel->Fd(), &event) < 0)
        {
            LOG(LogLevel::ERROR) << _epollfd << ": " << op << " updata " << channel->Fd() << " failure " << strerror(errno);
            abort();
        }
    }

    /// @brief 调用epoll_create
    /// @return
    void CreatPoller()
    {

        _epollfd = epoll_create(10);
        if (_epollfd < 0)
        {
            LOG(LogLevel::ERROR) << "creat epoll failed " << strerror(errno);
            abort();
        }
    }

public:
    Poller()
    {
        CreatPoller();
    }

    bool HasChannel(Channel *channel)
    {
        auto item = _channels.find(channel->Fd());
        return item != _channels.end();
    }

    /// @brief 对系统epoll_ctl调用的封装
    /// @param channel
    /// @param op
    void UpdateEvent(Channel *channel)
    {

        epoll_event event; // 创建epoll_ctl必须参数
        event.data.fd = channel->Fd();
        event.events = channel->Event();

        if (HasChannel(channel)) // 有记录改
        {
            Update(channel, EPOLL_CTL_MOD);
        }
        else // 没记录加
        {
            Update(channel, EPOLL_CTL_ADD);
            _channels.insert(std::make_pair(channel->Fd(), channel));
        }
    }

    /// @brief 对系统调用epoll_ctl的封装
    /// @param channel
    void RemoveEvent(Channel *channel)
    {
        if (HasChannel(channel))
        {
            Update(channel, EPOLL_CTL_DEL);
            LOG(LogLevel::DEBUG) << " RemoveEvent sucsess";
            _channels.erase(channel->Fd());
            LOG(LogLevel::DEBUG) << " Remove channel sucsess";
        }
        else
            LOG(LogLevel::WARNING) << "remove event not find fd " << channel->Fd(); // 没这个fd
    }

    /// @brief 进行非阻塞事件等待
    /// @param active
    /// @return
    int Poll(std::vector<Channel *> *active)
    {
        epoll_event events[MAX_EPOLL_SIZE];
        int num = 0;

        num = epoll_wait(_epollfd, events, MAX_EPOLL_SIZE, -1);
        // lg.logmessage(Debug, " in poll %d", num);
        if (num <= 0)
        {
            if (errno == EINTR)
                return 0;
            // lg.logmessage(Error, "poll epoll_wait error %s", strerror(errno));
            //  abort();
            LOG(LogLevel::ERROR) << "poll epoll_wait error " << strerror(errno);
            abort();
        }
        for (int i = 0; i < num; i++)
        {
            int fd = events[i].data.fd;
            if (_channels.find(fd) == _channels.end())
            {
                // lg.logmessage(Error, "poll not find fd %d", events[i].data.fd);
                LOG(LogLevel::ERROR) << "poll not find fd " << events[i].data.fd;
                abort();
            }
            _channels[fd]->SetTriggeredEvent(events[i].events); // 将被触发事件记录
            active->push_back(_channels[fd]);                   // 返回被触发id
            // 使其模块化仅仅就只有监控处理事件交给上层
        }
        return num;
    }
};

class EventLoop
{
private:
    using Func = std::function<void(void)>;
    std::vector<Func> _tasks;
    std::mutex _mutex;
    int _efd;
    Poller _poller;
    Channel _efd_cannel;

private:
    // void QueueInLOOP(const Func &fc)
    // {
    //     {
    //         std::lock_guard<std::mutex> lock(_mutex);
    //         _tasks.push_back(fc);
    //     }
    //     // 给eventfd写入一个数据，eventfd就会触发读事件
    //     WeakUpEventFd();
    // }

    void WeakUpEventFd()
    {
        int ret = eventfd_write(_efd, 1);
        if (ret < 0)
        {
            LOG(LogLevel::ERROR) << "eventfd_write failure";
            abort();
        }
    }

    int CreatEventfd()
    {
        _efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (_efd < 0)
        {
            LOG(LogLevel::ERROR) << "eventfd create failure";
            abort();
        }
        return _efd;
    }

    void HandleReadCallback()
    {
        eventfd_t value;
        eventfd_read(_efd, &value);
    }

    void RunAllTasks()
    {
        // LOG(LogLevel::DEBUG)<<"run tasks";
        std::vector<Func> tasks;
        {
            std::lock_guard<std::mutex> lock(_mutex);
            tasks.swap(_tasks); // 在多进程中将资源让出来
        }
        for (auto item : tasks)
        {
            item(); // 执行后续任务
        }
    }

public:
    EventLoop() : _efd_cannel(CreatEventfd(), *this)
    {
        _efd_cannel.SetReadCallback([this]()
                                    { this->HandleReadCallback(); });
        _efd_cannel.EnableRead();
    }
    void RunInLoop(Func fc)
    {
        fc();
    }

    void QueueInLOOP(const Func &fc)
    {
        {
            std::lock_guard<std::mutex> lock(_mutex);
            _tasks.push_back(fc);
        }
        // 给eventfd写入一个数据，eventfd就会触发读事件
        WeakUpEventFd();
    }

    // 由eventloop开始代替poller对channle接触
    void UpdateEvent(Channel *channel) { _poller.UpdateEvent(channel); }
    void RemoveEvent(Channel *channel) { _poller.RemoveEvent(channel); }
    void Start()
    {
        std::vector<Channel *> active;
        while (true)
        {
            _poller.Poll(&active);
            for (auto item : active)
            {
                item->HandleEvent(); // 收集信息为后面不妨碍多线程做准备(延迟销毁)
            }
            active.clear();
            RunAllTasks(); // 事件统一处理
        }
    }
};

class Connection;
typedef std::shared_ptr<Connection> ConnectionPtr;
/// @brief 连接关闭 正在连接 连接成功 关闭连接中
enum ConnectStatus
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
};
/// @brief 把一个连接结构话理应由socket和channel
class Connection : public std::enable_shared_from_this<Connection> ////却保从原始指针到倒多个共享的安全
{
    using CloseCallback = std::function<void(const ConnectionPtr &)>;
    using ReadCallback = std::function<void(const ConnectionPtr &, Buffer &in_buffer)>;

private:
    Socket _socket;
    int _fd;
    Channel _channel;        // 是不是用智能指针更好？(确保poller中的channels的正确释放指针应该就没问题其次更快)
    CloseCallback _close_cb; // 给上级调用者用当这一层要被释放时逐级往上释放资源
    ReadCallback _read_cb;   // 将收到的信息传给上一层让上层处理
    EventLoop &_event_loop;
    ConnectStatus _statu;
    // std::string _out_str;
    Buffer _out_buffer; // 从客户端来的信息
    Buffer _in_buffer;  // 服务端自己的信息

    // std::string _in_string;

private:
    void Close()
    {
        _statu = ConnectStatus::DISCONNECTED;
        _channel.Remove(); // 移除poller对channel的监控与管理
        if (_close_cb)
            _close_cb(shared_from_this()); // 移除上级Acceptor对自己的管理这里往后的代码直接不执行了
    }
    void CloseInLoop();

    void HandleClose()
    {
        Close();
    }

    void HandleRead()
    {
        // auto self = shared_from_this(); // 使得上层销毁不影响本体
        if (_statu != ConnectStatus::CONNECTED) // 如果链接失效则跳过
            return;
        char buf[InBufferSize];
        int ret = _socket.NoneBlockRecv(buf, InBufferSize);
        if (ret < 0)
        {
            LOG(LogLevel::WARNING) << "fd: " << _socket.Fd();
            Close();
            return;
        }
        _in_buffer.Write(buf, ret);
        // std::cout << buf << std::endl; // 这里应该是回调
        if (_read_cb != nullptr)
            _read_cb(shared_from_this(), _in_buffer);
    }

    void HandleWrite()
    {
        LOG(LogLevel::DEBUG) << "调用了";
        int ret = _socket.NoneBlockSend(_out_buffer.GetReadPostion(), _out_buffer.ReadableSzie());
        if (ret < 0)
        {

            // 出错时如果有读数据给它处理了没有就直接关了
            if (_in_buffer.ReadableSzie() > 0)
                _read_cb(shared_from_this(), _in_buffer);
            return Close();
        }
        _out_buffer.MoveReadPostion(ret);
        if (_out_buffer.ReadableSzie() == 0)
        {
            // 无数据则关闭写监控
            _channel.DisableWrite();
            if (_statu == ConnectStatus::DISCONNECTING)
            {
                return Close();
            }
        }
    }

    void Send(const std::string &date)
    {
        if (_statu == ConnectStatus::DISCONNECTED) // 如果处于断连则不做操作
            return;
        _out_buffer.Write(date);
        if (_channel.isWriteable() == false)
            _channel.EnableWrite();
    }

public:
    Connection(int fd, /*Poller &poller*/ EventLoop &event_loop) : _socket(fd),
                                                                   _channel(fd, event_loop),
                                                                   _fd(fd),
                                                                   _event_loop(event_loop),
                                                                   _statu(ConnectStatus::CONNECTED)
    {
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleClose, this));
        _channel.EnableRead();
    }

    void SetCloseCallback(const CloseCallback &cb) { _close_cb = cb; }
    void SetReadCallback(const ReadCallback &cb) { _read_cb = cb; }
    int Fd() const { return _fd; }
    void SendInLoop(const std::string &date)
    {
        _event_loop.QueueInLOOP([this, date]()
                                { this->Send(date); });
    };
};
class LoopThread
{
private:
    EventLoop *_loop;    // 将其在线程中创建实现逻辑上一个线程对应一个事件处理器
    std::thread _thread; // 创建一个线程
    std::mutex _mutex;   // 还未得到线程中的_loop之前需要锁住
    std::condition_variable _cond;

private:
    void ThreadEntry() // 线程入口函数创建event_loop并开始监听等待被使用
    {

        EventLoop loop;
        LOG(LogLevel::INFO) << gettid() << " loop thread created";
        {
            std::lock_guard<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all(); // 可以访问_loop了
        }
        loop.Start();
    }

public:
    LoopThread() : _loop(nullptr),
                   _thread([this]()
                           { this->ThreadEntry(); })
    {
    }
    LoopThread(LoopThread &&other) noexcept
        : _thread(std::move(other._thread)), _loop(other._loop)
    {
        other._loop = nullptr;
    }

    EventLoop *GetLoop()
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [this]()
                       { return this->_loop != nullptr; });
        }
        return _loop;
    }
};

/// @brief 带eventloop的线程池
class LoopThreadPool
{
private:
    std::vector<LoopThread> _threads;
    std::vector<EventLoop *> _loops;
    EventLoop *_base_loop;
    size_t _thread_count;
    size_t _next_thread; // 一个一个均衡负载
private:
    void CreatThreads()
    {
        _threads.resize(_thread_count);
        _loops.resize(_thread_count);
        for (int i = 0; i < _threads.size(); i++)
        {
            _loops.push_back(_threads[i].GetLoop());
        }
    }

public:
    // void Set

    EventLoop *NextThread()
    {
        if (_thread_count == 0)
            return _base_loop;
        _next_thread = (_next_thread + 1) % _thread_count; // 线程轮询
        return _threads[_next_thread].GetLoop();
    }

public:
    LoopThreadPool(EventLoop *baseloop, size_t thread_count = DEAFAULT_THREAD_COUNT) : _base_loop(baseloop), _thread_count(thread_count), _next_thread(0)
    {
        CreatThreads();
    }
};
class Acceptor
{
    using OnConnectedCallback = std::function<void(int)>;
    using OnClientreadCallback = std::function<void(const ConnectionPtr &, std::string)>;
    using OnClientcloseCallback = std::function<void(const ConnectionPtr &)>;

private:
    Socket _socket;
    Channel _channel;
    // Poller &_poller;
    //  std::vector<ConnectionPtr> _connections;
    EventLoop &_event_loop;
    LoopThreadPool &_threads;
    std::unordered_map<int, ConnectionPtr> _connections;
    OnConnectedCallback _on_connected_cb;

public:
    Acceptor(uint16_t port, EventLoop &event_loop, LoopThreadPool &threads) : _channel(CreateServer(port), event_loop), _event_loop(event_loop), _threads(threads)
    {
        _channel.SetReadCallback([this]()
                                 { this->HandleRead(); });
        _channel.EnableRead();
    }

    void SetOnconnectedCallback(const OnConnectedCallback &cb) { _on_connected_cb = cb; }

private:
    int CreateServer(uint16_t port)
    {
        if (_socket.CreatTcpServer(port))
            return _socket.Fd();
        else
            LOG(LogLevel::ERROR) << "create acceptor failure";
        abort();
    }
    /// @brief 在connections管理中删掉失效连接
    /// @param conn

    // void ClientClose(const ConnectionPtr conn)
    // {
    //     //_connections.erase(conn)
    //     LOG(LogLevel::DEBUG) << "HandleClientCloseCallback";
    //     auto ret = _connections.find(conn->Fd());
    //     if (ret != _connections.end())
    //     {
    //         _connections.erase(ret);
    //         LOG(LogLevel::DEBUG) << "remove conn from _connections success";
    //     }
    // }
    // void HandleClientCloseCallback(const ConnectionPtr conn);
    void HandleRead()
    {
        int fd = _socket.Accept();
        if (fd < 0)
        {
            LOG(LogLevel::ERROR) << "Accept failure";
            return;
        }
        LOG(LogLevel::INFO) << "accept a new link " << fd;
        // Connection *cli = new Connection(fd, _event_loop);
        // Connection *cli = new Connection(fd, *_threads.NextThread());
        // cli->SetCloseCallback([this](const auto &arg)
        //                       { this->HandleClientCloseCallback(arg); });
        // // cli->
        // _connections.insert(std::make_pair(fd, std::shared_ptr<Connection>(cli)));
        if (_on_connected_cb != nullptr)
            _on_connected_cb(fd);
    }
};

class TcpServer
{
private:
    EventLoop _base_loop;
    LoopThreadPool _threads;
    Acceptor _acceptor;
    std::unordered_map<int, ConnectionPtr> _connections;
    using OnMessageCallback = std::function<void(const ConnectionPtr &, Buffer &in_buffer)>;
    OnMessageCallback _on_Message_cb;

private:
    void HandleClientClose(const ConnectionPtr &conn)
    {
        _base_loop.QueueInLOOP([this, conn]()
                               { this->_connections.erase(conn->Fd()); });
    }
    void HandleOnConnected(int fd)
    {
        Connection *cli = new Connection(fd, *_threads.NextThread());
        cli->SetReadCallback(_on_Message_cb);
        cli->SetCloseCallback([this](const ConnectionPtr conn)
                              { this->HandleClientClose(conn); });
        _connections.insert(std::make_pair(fd, std::shared_ptr<Connection>(cli)));
    }

public:
    TcpServer(uint16_t port, size_t thraed_count = DEAFAULT_THREAD_COUNT) : _threads(&_base_loop, thraed_count), _acceptor(port, _base_loop, _threads)
    {
        //_acceptor.SetOnconnectedCallback();
        _acceptor.SetOnconnectedCallback([this](int fd)
                                         { this->HandleOnConnected(fd); });
    }

    void Start()
    {
        _base_loop.Start();
    }

    void SetOnMessageCallback(const OnMessageCallback &cb)
    {
        _on_Message_cb = cb;
    }
};

class EchoServer
{
private:
    TcpServer _tcp_serevr;

    void HandleOnMessage(const ConnectionPtr &conn, Buffer &in_buffer)
    {
        std::string out = "server: ";
        std::string str = in_buffer.ReadAsString(in_buffer.ReadableSzie(), true);
        out += str;
        conn->SendInLoop(out);
    }

public:
    EchoServer(uint16_t port = 8080) : _tcp_serevr(port)
    {
        _tcp_serevr.SetOnMessageCallback([this](const ConnectionPtr &conn, Buffer &in_buffer)
                                         { this->HandleOnMessage(conn, in_buffer); });
    }

    void Start()
    {
        _tcp_serevr.Start();
    }
};

void Channel::Update()
{ /*_poller.*/
    _event_loop.UpdateEvent(this);
}
void Channel::Remove() { /*_poller.*/ _event_loop.RemoveEvent(this); }

void Connection::CloseInLoop()
{
    _event_loop.QueueInLOOP([this]
                            { this->Close(); }); // 延迟销毁
}
// void Acceptor::HandleClientCloseCallback(const ConnectionPtr conn)
// {
//     _event_loop.QueueInLOOP([this, conn]()
//                             { this->ClientClose(conn); });
// }
